package com.ehl.capture.kafka;

public class KafkaMessageDivider
{
    private static int getInt(byte[] bb, int index)
    {
        return (bb[(index + 3)] & 0xFF) << 24 |
                (bb[(index + 2)] & 0xFF) << 16 |
                (bb[(index + 1)] & 0xFF) << 8 | (bb[(index + 0)] & 0xFF) << 0;
    }

    private static byte[] getBytes(byte[] b, int start, int length)
    {
        byte[] res = new byte[length];
        System.arraycopy(b, start, res, 0, length);
        return res;
    }

    private static byte[] getBytes(byte[] b, int start)
    {
        return getBytes(b, start, b.length - start);
    }

    public static byte[] getText(byte[] message)
    {
        int txtlen = getInt(message, 0);
        byte[] text = getBytes(message, 4, txtlen);
        return text;
    }

    public static byte[] getImage(byte[] message)
    {
        int txtlen = getInt(message, 0);
        byte[] image = getBytes(message, 4 + txtlen);
        return image;
    }

    public static byte[][] getMoreImage(byte[] message)
    {
        int len = getInt(message, 0);
        byte imgCount = message[(4 + len)];
        int curLen = 4 + len + 1;

        byte[][] images = new byte[imgCount][];

        for (byte i = 0; i < imgCount; i = (byte)(i + 1))
        {
            len = getInt(message, curLen);
            curLen += 4;
            images[i] = getBytes(message, curLen, len);
            curLen += len;
        }

        return images;
    }

    public static String getImageURL(String bpId, long offset)
    {
        return bpId + "/" + offset;
    }
}